package com.qys.livesMall.kafka.bean;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.bind.BindResult;
import org.springframework.boot.context.properties.bind.Bindable;
import org.springframework.boot.context.properties.bind.Binder;
import org.springframework.core.env.Environment;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.lang.reflect.Field;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

@Slf4j
@Component
public class EmallBeanFactoryAware implements BeanFactoryAware {

    @Autowired
    private Environment environment;

    private static final String SPRING_KAFKA_PREFIX = "spring.kafka";

    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        if (beanFactory instanceof DefaultListableBeanFactory) {
            DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) beanFactory;

            Binder binder = Binder.get(environment);
            //将YML中属性值映射到MAP中，后面根据配置前缀生成bean并注册到容器中，TODO 绑定可能有异常，加try catch稳一点
            BindResult<Map> bindResultWithPrefix = binder.bind(SPRING_KAFKA_PREFIX, Bindable.of(Map.class));
            if (!bindResultWithPrefix.isBound()) {
                return;
            }

            Map map = bindResultWithPrefix.get();
            Set set = map.keySet();
            Set<String> kafkaPropertyFiledNames = getKafkaPropertyFiledNames();

            //如果配置多个primary, 只设置第一个，TODO项目启动过程中，这个变量是否有并发问题
            boolean hasSetPrimary = false;
            //实例化每个带前缀的KafkaProperties、KafkaTemplate、
            for (Object object : set) {
                String prefix = object.toString();

                if (kafkaPropertyFiledNames.contains(prefix)) {
                    //不带前缀的正常配置忽略
                    continue;
                }

                String configPrefix = SPRING_KAFKA_PREFIX + "." + prefix;

                BindResult<KafkaProperties> kafkaPropertiesBindResult;
                try {
                    kafkaPropertiesBindResult = binder.bind(configPrefix, Bindable.of(KafkaProperties.class));
                    if (!kafkaPropertiesBindResult.isBound()) {
                        continue;
                    }
                } catch (Exception e) {
                    //一些配置不是在KafkaProperties属性，但是也不是前缀配置，在这一步会绑定失败，比如spring.kafka.topics配置,
                    //一些配置的名称是带-，KafkaProperties属性是驼峰，绑定是会出异常的，异常忽略
                    log.error("auto register kafka properties error, prefix is: {}", configPrefix);
                    continue;
                }

                //注册生产者（TODO 没配置生产者是否会报错）
                KafkaProperties kafkaProperties = kafkaPropertiesBindResult.get();
                String propertiesBeanName = prefix + "KafkaProperties";
                boolean isBeanExist = defaultListableBeanFactory.containsBean(propertiesBeanName);
                if (!isBeanExist) {
                    String primaryConfig = configPrefix + ".primary";
                    //没有默认的kafka配置，需要设置下primary
                    BindResult<Boolean> primaryBindResult = binder.bind(primaryConfig, Bindable.of(Boolean.class));
                    if (primaryBindResult.isBound() && primaryBindResult.get() && !hasSetPrimary) {
                        BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder.genericBeanDefinition(KafkaProperties.class);
                        defaultListableBeanFactory.registerBeanDefinition(propertiesBeanName, beanDefinitionBuilder.getBeanDefinition());
                        defaultListableBeanFactory.registerSingleton(propertiesBeanName, kafkaProperties);
                        defaultListableBeanFactory.getBeanDefinition(propertiesBeanName).setPrimary(true);
                        hasSetPrimary = true;
                    } else {
                        defaultListableBeanFactory.registerSingleton(propertiesBeanName, kafkaProperties);
                    }
                }

				//注册生产者KafkaTemplate
                String templateBeanName = prefix + "KafkaTemplate";
                if (!defaultListableBeanFactory.containsBean(templateBeanName)) {
                    KafkaTemplate kafkaTemplate = new KafkaTemplate<String, String>(
                            new DefaultKafkaProducerFactory<>(kafkaProperties.buildProducerProperties()));
                    defaultListableBeanFactory.registerSingleton(templateBeanName, kafkaTemplate);
                }

                String beanName = prefix + "KafkaListenerContainerFactory";
                if (!defaultListableBeanFactory.containsBean(beanName)) {
                    //注册消费者listener（TODO 没配置消费者是否会报错）
                    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                            new ConcurrentKafkaListenerContainerFactory<>();
                    factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(kafkaProperties.buildConsumerProperties()));
                    factory.setConcurrency(10);
                    factory.getContainerProperties().setPollTimeout(3000);
                    defaultListableBeanFactory.registerSingleton(beanName, factory);
                }
            }
        }
    }

    private static Set<String> getKafkaPropertyFiledNames () {
        Set<String> names = new HashSet<>();

        Field[] declaredFields = KafkaProperties.class.getDeclaredFields();
        if (declaredFields.length == 0) {
            return names;
        }

        for (Field declaredField : declaredFields) {
            names.add(declaredField.getName());
        }

        return names;
    }
}
